package com.k.activemq.openwire_sync1;

import com.k.activemq.openwire3.PubSubUtil;
import com.k.activemq.openwire_sync1.domain.ConnData;
import com.k.activemq.openwire_sync1.listener.CustomConsumerCallable;
import com.k.activemq.openwire_sync1.listener.CustomMessageListener;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.time.Duration;
import java.util.concurrent.*;

public class PubSync1 {
    public static void main(String[] args) {
        System.out.println("mq main start");
        // conn data
        String url = ConnData.url;
        String username = ConnData.username;
        String password = ConnData.password;
        String topic = ConnData.publishTopic;
        String topicReplay = ConnData.publishReplay;
        Boolean pubSubDomain = ConnData.pubSubDomain;
        Boolean persistentData = ConnData.PubConfig.persistentData;

        // conn
        ActiveMQConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        MessageConsumer consumer = null;
        try {
            // ConnectionFactory
            connectionFactory = new ActiveMQConnectionFactory(url);
            // Connection
            connection = connectionFactory.createConnection(username, password);
            connection.start();
            // Session
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 发送消息
            // 目的地
            Destination requestDestination = pubSubDomain ? session.createTopic(topic) : session.createQueue(topic);
            Destination responseDestination = pubSubDomain ? session.createTopic(topicReplay) : session.createQueue(topicReplay);

            // 发送者
            producer = session.createProducer(requestDestination);
            // 数据是否持久化(需要Broker 配置持久化 KaHaDB Mysql等 无特殊配置默认会使用 KaHaDB)
            if(persistentData){
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            }
            // 发送者 要发送的消息
            Message requestMsg = session.createTextMessage("{\"header\":{\"system_id\":\"\",\"subsystem_id\":\"\",\"msgid\":null,\"related_id\":null,\"send_time\":\"2023-09-21T11:08:55.8816136Z\",\"cmd\":\"Out_Status\",\"request\":\"PUC_SdsStatus\",\"request_type\":\"1\",\"reponse\":null,\"reponse_type\":null,\"action\":null},\"body\":{\"RemoteFilePath\":null,\"RemoteFilePath_27SP2\":null,\"FileName\":null,\"FileType\":null,\"FileSize\":null,\"TimeSpan\":null,\"Authgrant\":null,\"SapType\":null,\"Recipient\":{\"ID\":\"00001-924-data4puc6953\",\"Number\":\"data4puc6953\",\"Numbertype\":\"7\",\"PucId\":\"00001\",\"SystemId\":\"924\"},\"Sdscontent\":\"12345\",\"SDSType\":0,\"Sender\":{\"ID\":\"812A6DEA-55B2-416F-AB5C-083A2660AC57\",\"Number\":\"102\",\"Numbertype\":\"0\",\"PucId\":\"00001\",\"SystemId\":\"924\"},\"ProductName\":null}}");
            // 设置响应地址
            requestMsg.setJMSReplyTo(responseDestination);
            // 发送者-发送消息
            producer.send(requestMsg);
            System.out.println("server 发送消息完毕！");


            // 接收响应消息
            // 同步监听返回结果: 异步模拟同步方法
             consumer = session.createConsumer(responseDestination);
//            Long reqMills = System.currentTimeMillis();
//            consumer.setMessageListener(new CustomMessageListener(session,"101",reqMills,300L));

            // 阻塞方法
            ExecutorService executor = Executors.newSingleThreadExecutor();
            Future<Message> future = executor.submit(new CustomConsumerCallable(consumer));
            try {
                // 尝试在指定时间内获取结果，如果超时则抛出TimeoutException
                Message rspMsg =  future.get(6, TimeUnit.SECONDS);
                System.out.println("==== 获取执行结果：任务成功完成"+rspMsg);
            } catch (InterruptedException | ExecutionException e) {
                System.out.println("任务执行出错: " + e);
                Thread.currentThread().interrupt(); // 重新设置线程中断状态
            } catch (TimeoutException e) {
                System.out.println("任务超时，未在规定时间内完成");
                // 可以选择取消任务
//                future.cancel(true);
            } finally {
                System.out.println("=====finally");
                executor.shutdownNow(); // 关闭线程池
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            PubSubUtil.close(producer, consumer, session, connection);
        }
        System.out.println("mq main finish");
    }
}
